package com.informix.stream.cdc;

import com.informix.stream.api.IfmxStreamRecord;
import com.informix.stream.api.records.*;
import com.informix.stream.common.enums.IfmxStreamRecordType;
import com.informix.stream.common.exception.IfxStreamException;

import java.nio.ByteBuffer;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class IfxCDCRecordBuilder implements AutoCloseable {
    private final Logger logger = LoggerFactory.getLogger(IfxCDCRecordBuilder.class);
    private final Connection con;
    private final LinkedHashMap<String, LinkedHashMap<String, LinkedHashMap<String, String>>> columnMap = new LinkedHashMap<>();

    public IfxCDCRecordBuilder(Connection c) {
        this.con = c;
    }

    public IfmxStreamRecord buildRecord(byte[] bytes) throws SQLException, IfxStreamException {

        ByteBuffer buffer = ByteBuffer.wrap(bytes);
        int headerSize = buffer.getInt() - 16;
        int payloadSize = buffer.getInt();

        buffer.getInt();
        int recordType = buffer.getInt();
        byte[] header = new byte[headerSize];
        byte[] payload = new byte[payloadSize];
        buffer.get(header);
        buffer.get(payload);

        this.logger.trace("Record type [{}]", recordType);
        switch (recordType) {
            case 1:
                return new IfxCDCBeginTransactionRecord(header);
            case 2:
                return new IfxCDCCommitTransactionRecord(header);
            case 3:
                return new IfxCDCRollbackTransactionRecord(header);
            case 40:
                return new IfxCDCOperationRecord(
                        IfmxStreamRecordType.INSERT, header, payload, this.columnMap, this.con);
            case 41:
                return new IfxCDCOperationRecord(
                        IfmxStreamRecordType.DELETE, header, payload, this.columnMap, this.con);
            case 42:
                return new IfxCDCOperationRecord(
                        IfmxStreamRecordType.BEFORE_UPDATE, header, payload, this.columnMap, this.con);
            case 43:

                return new IfxCDCOperationRecord(
                        IfmxStreamRecordType.AFTER_UPDATE, header, payload, this.columnMap, this.con);
            case 119:
                return new IfxCDCTruncateRecord(header);
            case 200:
                // 记录表meta信息，后面解析insert/update/delete会用到
                ByteBuffer meta200 = ByteBuffer.wrap(bytes);
                byte[] meta200header = new byte[headerSize + payloadSize - 1];
                meta200.position(16);//跳过公共报文
                meta200.get(meta200header);
                ByteBuffer ggg = ByteBuffer.wrap(meta200header);
                CDCMetadataRecord metaRecode200 = new CDCMetadataRecord(ggg);
                LinkedHashMap<String, LinkedHashMap<String, String>> columns = new LinkedHashMap<>();
                for (String colName : metaRecode200.getColNames()) {
                    LinkedHashMap<String, String> colInfo = new LinkedHashMap<>();
                    colInfo.put("colType", metaRecode200.getColTypes().get(colName));
                    if (metaRecode200.getColLengths().get(colName) != null) {
                        colInfo.put("colLength", metaRecode200.getColLengths().get(colName).toString());
                    }
                    if (metaRecode200.getColPrecisions().get(colName) != null) {
                        colInfo.put("colPrecision", metaRecode200.getColPrecisions().get(colName).toString());
                    }

                    columns.put(colName, colInfo);
                }

                IfxCDCMetaDataRecord record = new IfxCDCMetaDataRecord(columns, header);
                this.columnMap.put(record.getLabel(), columns);
                return record;
            case 201:
                return new IfxCDCTimeoutRecord(header);
            case 202:
                return new IfxCDCErrorRecord(header);
            default:
                throw new IfxStreamException("Record type:" + recordType + " not defined.");
        }
    }

    @Override
    public void close() throws SQLException {
        this.con.close();
    }
}
